package com.shujia.flink.source;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.connector.file.src.FileSource;
import org.apache.flink.connector.file.src.reader.TextLineInputFormat;
import org.apache.flink.core.fs.Path;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.time.Duration;

public class Demo1FIleSource {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //1、读取文件(老版本)  -- 有界流
        DataStream<String> studentDS = env.readTextFile("flink/data/students.txt");

        //studentDS.print();


        /*
         *2、新版本读取文件的方式
         * 1、可以有界流读取文件、
         * 2、可以无界流读取文件，增加定时监控参数
         */
        FileSource<String> fileSource = FileSource
                .forRecordStreamFormat(
                        new TextLineInputFormat(),//读取文件格式
                        new Path("flink/data/word")//文件的路径
                )
                //定时监控目录读取目录下新的文件。得到一个无界流
                .monitorContinuously(Duration.ofSeconds(5))
                .build();

        //使用file source
        DataStreamSource<String> wordsDS = env
                .fromSource(fileSource, WatermarkStrategy.noWatermarks(), "fileSource");

        wordsDS.print();

        env.execute();


    }
}
